import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import java.io.IOException;

/**
 * @author legolas
 * @date 2020/3/13 下午7:06
 * 需求：读取hdfs上的hello.txt文件，计算文件中每个单词出现的总次数
 * hello.txt:
 * hello you
 * hello me
 * <p>
 * map处理后的结果：形式如下
 * hello 1
 * hello 1
 * me 1
 * you 1
 * <p>
 * reduce处理后的结果：形式如下
 * hello 2
 * me 1
 * you 1
 */

public class WordCountJob {

    /**
     * 创建自定义Mapper类
     * 这里不能使用java的数据类型，必须使用hadoop提供的数据类型
     * <p>
     * k1：LongWritable，数字类型,数字类型每行行首的偏移量
     * v1:Text:v1字符串类型
     * k2：Text，key为分类后的单词
     * v2：LongWritable，出现次数
     */
    public static class MyMapper extends Mapper<LongWritable, Text, Text, LongWritable> {

        /**
         * 需要实现的map函数
         * 这个map函数就是可以接收k1，v1   产生k2，v2
         *
         * @param k1
         * @param v1
         * @param context
         * @throws IOException
         * @throws InterruptedException
         */

        @Override
        protected void map(LongWritable k1, Text v1, Context context) throws IOException, InterruptedException {
            //k1代表的是每一行的行首偏移量，v1代表的时每一行的内容
            //对获取到的每一行数据进行切割，把单词切割出来
            String[] words = v1.toString().split(" ");
            //迭代切割出来的单词数据
            for (String word : words) {
                //把迭代出来的单词封装成k2，v2的形式
                Text k2 = new Text(word);
                LongWritable v2 = new LongWritable(1L);
                System.out.println("k2:" + k2.toString() + "....v2:" + v2.toString());
                //把<k2,v2>写出去
                context.write(k2, v2);
            }
        }
    }

    /**
     * 创建自定义reducer类
     */
    public static class MyReduce extends Reducer<Text, LongWritable, Text, LongWritable> {
        /**
         * 针对v2s的数据进行累加求和，并且最终把数据转化伪k3,v3写出去
         *
         * @param k2
         * @param v2s
         * @param context
         * @throws IOException
         * @throws InterruptedException
         */
        @Override
        protected void reduce(Text k2, Iterable<LongWritable> v2s, Context context) throws IOException, InterruptedException {
            //创建一个sum变量，保存v2s的和
            long sum = 0L;
            for (LongWritable v2 : v2s) {
                sum += v2.get();
            }
            //组装k3,v3
            Text k3 = k2;
            LongWritable v3 = new LongWritable(sum);
            System.out.println("k3:" + k3.toString() + ".....v3:" + v3.toString());
            context.write(k3, v3);
        }
    }

    /**
     * 组装job： map+reduce
     *
     * @param args
     */
    public static void main(String[] args) {
        try {
            if (args.length != 2) {
                //如果参数不够，程序直接退出
                System.out.print("请指定输入路径和输出路径！");
                System.exit(100);
            }
            //创建Job需要的配置参数
            Configuration conf = new Configuration();
            //创建一个job
            Job job = Job.getInstance(conf);

            //注意：这一行必须设置，否则在集群中执行找不到WordCountJob这个类
            job.setJarByClass(WordCountJob.class);

            //指定输入路径
            FileInputFormat.setInputPaths(job, new Path(args[0]));
            FileOutputFormat.setOutputPath(job, new Path(args[1]));
            //指定map相关代码
            job.setMapperClass(MyMapper.class);
            job.setMapOutputKeyClass(Text.class);
            job.setMapOutputValueClass(LongWritable.class);

            //指定reduce相关代码
            job.setReducerClass(MyReduce.class);
            job.setOutputKeyClass(Text.class);
            job.setOutputValueClass(LongWritable.class);

            //提交job
            job.waitForCompletion(true);

        } catch (Exception e) {
            e.printStackTrace();
        }
    }


}
